package com.hexiang.rocketmq.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@RocketMQMessageListener(topic = "test_topic", consumerGroup = "mock-reconsume-by-exception-spring-consumer-group", selectorExpression = "*", selectorType = SelectorType.TAG)
public class MockReConsumeByExceptionRocketMQListener implements RocketMQListener<MessageExt> {

    private static final Integer MAX_RETRY_TIMES = 2;

    @Override
    public void onMessage(MessageExt messageExt) {

        try {
            //逐条消费
            String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
            log.info("Message Consumer: Handle New Message: messageId: " + messageExt.getMsgId() + ",topic: " +
                    messageExt.getTopic() + ",tags: " + messageExt.getTags() + ",messageBody: " + messageBody);
            //模拟业务异常
            int i = 1 / 0;
        } catch (Exception e) {
            if (messageExt.getReconsumeTimes() < MAX_RETRY_TIMES) {
                throw new RuntimeException(e);
            }
        }
    }

}